/** pipedstream.d - Provided a buffer-less pipe model for streams like the Unix's pipe. 

  Written in the D programming language 1.0

Authors:	Wei Li <oldrev@gmail.com>
Copyright:	Copyright (c) 2007 by Wei Li. All rights revserved.
License:	BSD
*/


module dotmars.io.pipe;

import dotmars.io.stream;
import dotmars.threading.semaphore;
import dotmars.base.math;

/**
  It is used to link two threads, especially ICoder.
Examples:
--------------------------
FileStream ins = new FileStream(...);
FileStream outs = new FileStream(...);
auto pin = new PipedInputStream;
auto pout = new PipedOutputStream(pin);
ICoder compressor = new GZipCoder;
ICoder cryptor = new AES256Coder;
auto proc1 = new Thread(&compressor.code(ins, pout));
auto proc2 = new Thread(&cryptor.code(pin, outs));
proc1.run();
proc2.run();
proc2.join(); // Waitting for last thread (proc2) to complete
--------------------------
*/

class PipedInputStream : Stream
{
	private ubyte* m_buffer;
	private size_t m_available;
	private Semaphore m_fullEvent;
	private Semaphore m_emptyEvent;
	private bool m_closed;

	public this()
	{
		m_available = 0;
		m_buffer = null;

		m_fullEvent = new Semaphore(1);
		m_emptyEvent = new Semaphore(1);

		m_closed = false;
	}

	public synchronized override void close()
	{
		if(!m_closed)
		{
			m_closed = true;
			m_fullEvent.unlock();
		}
	}

	// This method will be called by producer thread
	private void receive(void* data, size_t count)
	{
		assert(!m_closed);
		assert(m_available == 0);

		if(count == 0) return;

		m_buffer = cast(ubyte*)data;
		m_available = count;
		m_fullEvent.unlock();
		m_emptyEvent.lock();
	}

	private size_t read(void* data, size_t count)
	{
		assert(!m_closed);

		if(count == 0)
			return 0;

		size_t rsize = count;
		size_t ret = 0;
		ubyte* buf = cast(ubyte*)data;

		while(rsize > 0)
		{
			if(m_available == 0)
				m_fullEvent.lock();

			if(m_available != 0) 
			{
				size_t rc = Math.min(rsize, m_available);
				buf[0..rc] = m_buffer[0..rc];
				buf += rc;
				ret += rc;
				m_buffer += rc;
				m_available -= rc;
				rsize -= rc;
			}

			if(m_available == 0)
			{
				m_emptyEvent.unlock();

				if(m_closed)
					return ret;
			}

		}

		return ret;
	}

	public size_t available()
	{
		return m_available;
	}


	public override bool readable() { return true; }
	public override bool writable() { return false; }
	public override bool seekable() { return false; }

	public override void write(void* data, size_t count)
	{
		assert(false, "Not supported");
	}

	public override long seek(long offset, SeekOrigin origin) {
		assert(false, "Not supported");
		return 0L;
	}

	public override void length(long len) {
		assert(false, "Not supported");
	}

	public override long length() {
		assert(false, "Not supported");
		return 0L;
	}

	public override void flush()
	{
		//do nothing
	}
}


class PipedOutputStream : Stream
{
	private PipedInputStream m_sink = null;
	private bool m_closed;

	public this(PipedInputStream sink)
	in {
		assert(sink !is null);
	}
	body {
		m_sink = sink;
		m_closed = false;
	}

	public override bool readable() { return false; }
	public override bool writable() { return true; }
	public override bool seekable() { return false; }

	public override size_t read(void* data, size_t count)
	{
		assert(false, "Not supported");
		return 0;
	}

	public override void write(void* data, size_t count)
	{
		assert(!m_closed);
		m_sink.receive(data, count);
	}

	public override long seek(long offset, SeekOrigin origin) {
		assert(false, "Not supported");
		return 0L;
	}

	public override void length(long len) {
		assert(false, "Not supported");
	}

	public override long length() {
		assert(false, "Not supported");
		return 0L;
	}

	public override void flush()
	{
		//do nothing
	}

	public synchronized override void close()
	{
		if(!m_closed)
		{
			m_closed = true;
		}
	}

} //class OutputStream
